package org.trade.common.rocketmq;

import java.util.List;

import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

/*
 * 作为consumer中的监听器，
 */
public class AceMessageListener implements MessageListenerConcurrently {

	private IMessageProcessor processor;

	public void setProcessor(IMessageProcessor processor) {
		this.processor = processor;
	}

	//处理信息
	public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
		for (MessageExt msg : msgs) {
			boolean result = processor.handleMessage(msg);
			if(!result) {
				return ConsumeConcurrentlyStatus.RECONSUME_LATER;
			}
		}
		return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
	}

}
